from pykafka import KafkaClient,SslConfig
from pykafka.exceptions import SocketDisconnectedError
import Queue

def get_client():
    return KafkaClient(hosts="127.0.0.1:9092",)
def get_topic(clienta):
    return clienta.topics
def produce_syn(topic):
    with topic.get_sync_producer() as producer:
        for i in range(4):
            try:
                producer.produce('test message ')
            except Exception as e:
                print 'fail'
                print e.message
def produce_asyn(topic):
    with topic.get_producer(delivery_reports=True) as producer:
        count = 0
        while True:
            count += 1
            producer.produce('test msg', partition_key='{}'.format(count))
            if count % 10 ** 5 == 0:  # adjust this or bring lots of RAM ;)
                while True:
                    try:
                        msg, exc = producer.get_delivery_report(block=False)
                        if exc is not None:
                            print 'Failed to deliver msg {}: {}'.format(msg.partition_key, repr(exc))
                        else:
                            print 'Successfully delivered msg {}'.format(msg.partition_key)
                    except Queue.Empty:
                            break
def consumer_simple(topic):
    consumer=topic.get_simple_consumer(auto_commit_enable=False)
    for message in consumer:
        if message is not None:
            print message.offset,message.value
def consumer_balanced(topic):
    consumer = None
    try:
        consumer = topic.get_balanced_consumer(consumer_group='test-group',
                                               zookeeper_connect='127.0.0.1:2181',
                                               auto_commit_enable=True,
                                               auto_commit_interval_ms=1000)
        while True:
            message = consumer.consume()
            print message.__dict__
    except SocketDisconnectedError:
        print 'handle error here'
    finally:
        if consumer:
            consumer.stop()
def control():
    clienta=get_client()
    print  get_topic(clienta=clienta)
    topic= get_topic(clienta=clienta)['test']
    print topic
    # produce_asyn(topic=topic)
    consumer_simple(topic=topic)
    consumer_balanced(topic=topic)

if __name__=="__main__":
    # control()
    client=KafkaClient(hosts="127.0.0.1:9092",)
    topica= client.topics['test']
    # producer=topica.get_producer()
    # print producer
    # res=producer.produce('test message 4')
    # print res
    # consumer=topica.get_simple_consumer()
    # for i in consumer:
    #     if i is not None:
    #         print i.offset,i.value

    blance_cons=topica.get_balanced_consumer(consumer_group='testgroup',auto_commit_enable=True,zookeeper_connect='127.0.0.1:2181')
    print blance_cons
    for i in blance_cons:
        print i.value